package seckill

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"mall/common/mqueuex"
	// "mall/service/product/rpc/productclient"
	"mall/service/promotion/rpc/internal/svc"
	"mall/service/promotion/rpc/types/promotion"

	"github.com/sirupsen/logrus"
	"github.com/zeromicro/go-zero/core/collection"
	"github.com/zeromicro/go-zero/core/limit"
	"github.com/zeromicro/go-zero/core/logx"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

const (
	limitPeriod       = 10
	limitQuota        = 1
	seckillUserPrefix = "seckill#u#"
	localCacheExpire  = time.Second * 60

	batcherSize     = 100
	batcherBuffer   = 100
	batcherWorker   = 10
	batcherInterval = time.Second
)

type CreateSeckillOrderLogic struct {
	ctx        context.Context
	svcCtx     *svc.ServiceContext
	limiter    *limit.PeriodLimit
	localCache *collection.Cache
	batcher    *mqueuex.Batcher
	logx.Logger
}

func NewCreateSeckillOrderLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CreateSeckillOrderLogic {
	localCache, err := collection.NewCache(localCacheExpire)
	if err != nil {
		logx.Errorw("local cache init error", logx.Field("detail", err.Error()))
		return nil
	}

	l := &CreateSeckillOrderLogic{
		ctx:        ctx,
		svcCtx:     svcCtx,
		Logger:     logx.WithContext(ctx),
		localCache: localCache,
		limiter:    limit.NewPeriodLimit(limitPeriod, limitQuota, svcCtx.BizRedis, seckillUserPrefix),
	}

	b := mqueuex.NewBatcher(
		mqueuex.WithSize(batcherSize),
		mqueuex.WithBuffer(batcherBuffer),
		mqueuex.WithWorker(batcherWorker),
		mqueuex.WithInterval(batcherInterval),
	)
	b.Sharding = func(key string) int {
		pid, _ := strconv.ParseInt(key, 10, 64)
		return int(pid) % batcherWorker
	}
	b.Do = func(ctx context.Context, val map[string][]interface{}) {
		var msgs []*mqueuex.SeckillMessage
		for _, vs := range val {
			for _, v := range vs {
				msgs = append(msgs, v.(*mqueuex.SeckillMessage))
			}
		}
		kd, err := json.Marshal(msgs)
		if err != nil {
			logrus.Info(fmt.Sprintf("Batcher.Do json.Marshal msgs: %v, error: %v", msgs, err))
		} else {
			if err = svcCtx.KqClient.Push(string(kd)); err != nil {
				logrus.Info(fmt.Sprintf("KqClient.Push kd: %s error: %v", string(kd), err))
			}
		}
	}
	l.batcher = b
	l.batcher.Start()
	return l
}

func (l *CreateSeckillOrderLogic) CreateSeckillOrder(in *promotion.SeckillOrderReq) (*promotion.BaseResp, error) {
	code, _ := l.limiter.Take(strconv.FormatInt(in.UserId, 10))
	if code == limit.OverQuota {
		return nil, status.Errorf(codes.OutOfRange, "Number of requests exceeded the limit")
	}

	// p, err := l.svcCtx.ProductRPC.Product(l.ctx, &productclient.ProductItemRequest{ProductId: in.ProductId})
	// if err != nil {
	// 	return nil, err
	// }

	// if p.Stock <= 0 {
	// 	return nil, status.Errorf(codes.OutOfRange, "Insufficient stock")
	// }
	err := l.batcher.Add(strconv.FormatInt(in.VariationId, 10), &mqueuex.SeckillMessage{Uid: uint64(in.UserId), Vid: uint64(in.VariationId)})
	if err != nil {
		logrus.Info(fmt.Sprintf("l.batcher.Add uid: %d vid: %d error: %v", in.UserId, in.VariationId, err))
		return nil, err
	}

	return &promotion.BaseResp{}, nil
}
